package com.we.mexico.flink;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import com.alibaba.fastjson.JSONObject;
import flink.utils.AppConfig;
import flink.utils.FlinkUtils;

import java.util.Map;

public class Test {

    static final ParameterTool init = AppConfig.init("test.properties");

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //  kafka source
        final Map<String, DataStream<JSONObject>> dataStreamMap =
                FlinkUtils.addKafkaSource(env, null);
        // 这里是自己的代码逻辑
        final DataStream<JSONObject> fdm_hhl01_user_pid = dataStreamMap.get("fdm_hhl01_user_pid");
        // kafka sink
        fdm_hhl01_user_pid.sinkTo(FlinkUtils.addKafkaSink());
        env.execute();
    }
}
